package org.raymond.iworks.study.mq.kafka.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.raymond.iworks.study.mq.kafka.common.Bar2;
import org.raymond.iworks.study.mq.kafka.common.Foo2;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.TopicBuilder;
import org.springframework.kafka.core.*;
import org.springframework.kafka.listener.CommonErrorHandler;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.kafka.support.converter.JsonMessageConverter;
import org.springframework.kafka.support.converter.RecordMessageConverter;
import org.springframework.kafka.support.mapping.DefaultJackson2JavaTypeMapper;
import org.springframework.kafka.support.mapping.Jackson2JavaTypeMapper;
import org.springframework.kafka.support.serializer.JsonSerializer;
import org.springframework.util.backoff.FixedBackOff;

import java.util.HashMap;
import java.util.Map;

/**
 * @author raymond
 * @version V1.0
 * @Description: TODO
 */
@Slf4j
@Configuration
@ConditionalOnProperty(name="kafka.sample", havingValue="demo7")
public class Demo7Config {
    private static final String KAFKA_SERVER_ADDR = "192.168.0.125:9094,192.168.0.125:9194,192.168.0.125:9294";

    @Bean
    public KafkaAdmin admin() {
        Map<String, Object> configs = new HashMap<>();
        configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_SERVER_ADDR);
        return new KafkaAdmin(configs);
    }
    @Bean
    public NewTopic topicFoo() {
        log.info("topic demo7-foo");
        return TopicBuilder.name("demo7-foo")
                .partitions(1)
                .replicas(1)
                .build();
    }
    @Bean
    public NewTopic topicBar() {
        log.info("topic demo7-bar");
        return TopicBuilder.name("demo7-bar")
                .partitions(1)
                .replicas(1)
                .build();
    }

    @Bean
    public CommonErrorHandler errorHandler(KafkaOperations<Object, Object> template) {
        return new DefaultErrorHandler(
                new DeadLetterPublishingRecoverer(template), new FixedBackOff(1000L, 2));
    }

    public ProducerFactory producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

    @Bean
    public Map<String, Object> producerConfigs() {
        Map<String, Object> props = Demo1Config.producerConfigs(JsonSerializer.class);
        props.put(JsonSerializer.TYPE_MAPPINGS,
                "foo:org.raymond.icloud.modules.kafka.common.Foo1," +
                        "bar:org.raymond.icloud.modules.kafka.common.Bar1");
        return props;
    }

    @Bean
    public RecordMessageConverter converter() {
        JsonMessageConverter converter = new JsonMessageConverter();
        DefaultJackson2JavaTypeMapper typeMapper = new DefaultJackson2JavaTypeMapper();
        typeMapper.setTypePrecedence(Jackson2JavaTypeMapper.TypePrecedence.TYPE_ID);
        typeMapper.addTrustedPackages("org.raymond.icloud.modules.kafka.common");
        Map<String, Class<?>> mappings = new HashMap<>();
        mappings.put("foo", Foo2.class);
        mappings.put("bar", Bar2.class);
        typeMapper.setIdClassMapping(mappings);
        converter.setTypeMapper(typeMapper);
        return converter;
    }

    @Bean
    public KafkaTemplate kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }

    @Bean
    public ConsumerFactory consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }

    @Bean
    public Map<String, Object> consumerConfigs(){
        return Demo1Config.consumerConfigs(ByteArrayDeserializer.class);
    }
}
